package com.shujia.traffic.dws

import com.shujia.traffic.common.SparkUtil
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql._

import scala.io.{BufferedSource, Source}

object DwsAppVioZqzfDay extends SparkUtil{
  /**
   * SparkSQL任务提交的命令：
   * spark-submit --master local --conf spark.sql.shuffle.partitions=2 --class com.shujia.traffic.dws.DwsAppVioZqzfDay traffic-1.0-SNAPSHOT.jar 20250425
   */

  override def run(): Unit = {
    val spark: SparkSession = sparkSession

    import org.apache.spark.sql.functions._
    import spark.implicits._

    // 加载dwd层事故事实表的数据
    val acdDF: DataFrame = spark.table("dwd.dwd_base_acd_file_msk_d").where($"pt" === day)
    val acdHumanDF: DataFrame = spark.table("dwd.dwd_base_acd_filehuman_msk_d").where($"pt" === day)
    val forceDF: DataFrame = spark.table("dwd.dwd_base_vio_force_msk_d").where($"pt" === day)
    val surveilDF: DataFrame = spark.table("dwd.dwd_base_vio_surveil_msk_d").where($"pt" === day)

    val acdJoinDF: DataFrame = acdHumanDF
      .select($"sgbh", $"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")

    val acdVioDF: DataFrame = acdDF
      .select($"sgbh", $"sgfssj" as "wfsj")
      .join(acdJoinDF, List("sgbh"), "inner")
      .select($"wfsj", explode(array($"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")) as "wfxw")

    val forceVioDF: DataFrame = forceDF
      .select($"wfsj", explode(array($"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")) as "wfxw")

    // 加载wfkj文件，将不同的违法口径解析成KV格式，构建Map
    //    val bs: BufferedSource = Source.fromFile("traffic/data/wfkj.txt")

    // maven打jar包时只会将resources下的文件打到jar包中，其他位置的文件默认会忽略，故无法使用路径直接加载文件
    // 可以考虑将数据放入远程数据库中，或者将数据文件打入jar包
    val bs: BufferedSource = Source
      .fromInputStream(
        this
          // 获取当前运行的类
          .getClass
          // 获取类加载器
          .getClassLoader
          // 获取jar包根目录
          .getResourceAsStream("wfkj.txt")
      )

    val wfkjMap: Map[String, List[String]] = bs.getLines()
      .map(line => {
        val splits: Array[String] = line.split("=")
        val index: String = splits(0)
        val codeList: List[String] = splits(1).split(",").toList
        (index, codeList)
      }).toMap

    bs.close()

    val unionDF: Dataset[Row] = surveilDF
      .select($"wfsj", $"wfxw")
      .unionAll(forceVioDF)
      .unionAll(acdVioDF)
      // 过滤掉违法行为空的记录
      .where(trim($"wfxw") =!= "" and upper(trim($"wfxw")) =!= "NULL" and upper(trim($"wfxw")) =!= "\\N")

    var drCntDF: DataFrame = unionDF
      // 统计每日的不同类别的违法行为
      .groupBy(substring($"wfsj", 1, 10) as "wfsj")
      .agg(
        count("*") as "dr_zd_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("zjjs", Nil)), 1).otherwise(0)) as "dr_zjjs_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("yjjs", Nil)), 1).otherwise(0)) as "dr_yjjs_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("zdhp", Nil)), 1).otherwise(0)) as "dr_zdhp_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wzbz", Nil)), 1).otherwise(0)) as "dr_wzbz_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("tpjp", Nil)), 1).otherwise(0)) as "dr_tpjp_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wdtk", Nil)), 1).otherwise(0)) as "dr_wdtk_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("jspz", Nil)), 1).otherwise(0)) as "dr_jspz_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wxjs", Nil)), 1).otherwise(0)) as "dr_wxjs_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("cy", Nil)), 1).otherwise(0)) as "dr_cy_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("cz", Nil)), 1).otherwise(0)) as "dr_cz_wfs"
      )

    var dyCntDF: DataFrame = unionDF
      // 统计每月的不同类别的违法行为
      .groupBy(substring($"wfsj", 1, 7) as "wfny")
      .agg(
        count("*") as "dy_zd_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("zjjs", Nil)), 1).otherwise(0)) as "dy_zjjs_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("yjjs", Nil)), 1).otherwise(0)) as "dy_yjjs_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("zdhp", Nil)), 1).otherwise(0)) as "dy_zdhp_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wzbz", Nil)), 1).otherwise(0)) as "dy_wzbz_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("tpjp", Nil)), 1).otherwise(0)) as "dy_tpjp_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wdtk", Nil)), 1).otherwise(0)) as "dy_wdtk_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("jspz", Nil)), 1).otherwise(0)) as "dy_jspz_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wxjs", Nil)), 1).otherwise(0)) as "dy_wxjs_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("cy", Nil)), 1).otherwise(0)) as "dy_cy_wfs"
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("cz", Nil)), 1).otherwise(0)) as "dy_cz_wfs"
      )

    // 使用循环分别统计不同违法类别今年的违法数及同比
    val indexList: List[String] = List[String]("zd", "zjjs", "yjjs", "zdhp", "wzbz", "tpjp", "wdtk", "jspz", "wxjs", "cy", "cz")
    indexList.foreach(index => {
      drCntDF = drCntDF
        .withColumn(s"jn_${index}_wfs", sum($"dr_${index}_wfs") over Window.partitionBy(year($"wfsj")))

      dyCntDF = dyCntDF
        .withColumn(s"last_dy_${index}_wfs", lag($"dy_${index}_wfs", 1) over Window.partitionBy(month($"wfny")).orderBy(year($"wfny")))
        .withColumn(s"${index}_tb", concat(round(coalesce($"dy_${index}_wfs" / $"last_dy_${index}_wfs" - 1, expr("0")) * 100, 2), expr("'%'")))
    })

    // 创建对应结果表的分区
    spark.sql(
      s"""
         |alter table dws.dws_app_vio_zqzf_d add if not exists partition (pt = '${day}')
         |""".stripMargin)

    drCntDF
      .join(dyCntDF, substring($"wfsj", 1, 7) === $"wfny", "left")
      .select(
        $"wfsj"
        , $"dr_zd_wfs"
        , $"jn_zd_wfs"
        , $"zd_tb"
        , $"dr_zjjs_wfs"
        , $"jn_zjjs_wfs"
        , $"zjjs_tb"
        , $"dr_yjjs_wfs"
        , $"jn_yjjs_wfs"
        , $"yjjs_tb"
        , $"dr_zdhp_wfs"
        , $"jn_zdhp_wfs"
        , $"zdhp_tb"
        , $"dr_wzbz_wfs"
        , $"jn_wzbz_wfs"
        , $"wzbz_tb"
        , $"dr_tpjp_wfs"
        , $"jn_tpjp_wfs"
        , $"tpjp_tb"
        , $"dr_wdtk_wfs"
        , $"jn_wdtk_wfs"
        , $"wdtk_tb"
        , $"dr_jspz_wfs"
        , $"jn_jspz_wfs"
        , $"jspz_tb"
        , $"dr_wxjs_wfs"
        , $"jn_wxjs_wfs"
        , $"wxjs_tb"
        , $"dr_cy_wfs"
        , $"jn_cy_wfs"
        , $"cy_tb"
        , $"dr_cz_wfs"
        , $"jn_cz_wfs"
        , $"cz_tb"
      )
      .write
      .format("csv")
      .option("sep", "|")
      .mode(SaveMode.Overwrite)
      // 只会自动创建目录，不会创建分区
      .save(s"/project/traffic/dws/dws_app_vio_zqzf_d/pt=${day}")

  }
}
